package com.lianda.example.hot;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.io.PojoCsvInputFormat;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.io.File;
import java.net.URL;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

//热门商品
//http://wuchong.me/blog/2018/11/07/use-flink-calculate-hot-items/
public class HotItems {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设定时间模式
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        // UserBehavior.csv 的本地文件路径
        URL fileUrl = HotItems.class.getClassLoader().getResource("data/UserBehavior.csv");
        Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));

        // 抽取 UserBehavior 的 TypeInformation，是一个 PojoTypeInfo
        PojoTypeInfo<UserBehavior> pojoType = (PojoTypeInfo<UserBehavior>) TypeExtractor
                .createTypeInfo(UserBehavior.class);

        //确定字段反射顺序
        String[] fieldOrder = new String[] {"userId", "itemId", "categoryId", "behavior", "timestamp"};

        PojoCsvInputFormat<UserBehavior> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);

        //用csv文件创建数据源
        DataStream<UserBehavior> dataSource = env.createInput(csvInput, pojoType);


        //每隔5分钟输出过去一小时内点击量最多的前 N 个商品

        //获取业务时间，且生产Watermark(水位)
        DataStream<UserBehavior> timedData = dataSource.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
            @Override
            public long extractAscendingTimestamp(UserBehavior element) {
                //转换成毫秒
                return element.timestamp * 1000;
            }
        });

        //过滤点击事件
        DataStream<UserBehavior> pvData = timedData.filter(new FilterFunction<UserBehavior>() {
            @Override
            public boolean filter(UserBehavior userBehavior) throws Exception {
                return userBehavior.behavior.equals("pv");
            }
        });

        //窗口统计点击量
        //每隔5分钟统计一次
        //滑动窗口
        DataStream<ItemViewCount> windowedData = pvData.keyBy("itemId")
                // 最近1小时，每隔5分钟滑动窗口
                .timeWindow(Time.minutes(60), Time.minutes(5))
                //使用AggregateFunction提前聚合掉数据，减少 state 的存储压力
                .aggregate(new CountAgg(), new WindowResultFunction()); // 聚合出当前商品在时间窗口内的统计数量

        DataStream<String> topItems = windowedData
                .keyBy("windowEnd")
                .process(new TopNHotItems(3));//求点击量前三名

        topItems.print();

        env.execute("Hot Items Job");

    }

    //介绍Function：https://www.dazhuanlan.com/2020/03/09/5e65b1f7c7f1d/
    // https://cloud.tencent.com/developer/article/1481829
    public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String> {

        private final int topSize;

        public TopNHotItems(int topSize) {
            this.topSize = topSize;
        }
        // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算
        //使用State，保证在发生故障时，状态数据的不丢失和一致性
        //做到了 exactly-once 的语义保证
        private ListState<ItemViewCount> itemState;

        //初始化状态
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //状态注册
            ListStateDescriptor<ItemViewCount> itemsStateDesc = new ListStateDescriptor<ItemViewCount>(
                    "itemState-state",
                    ItemViewCount.class
            );
            itemState = getRuntimeContext().getListState(itemsStateDesc);
        }

        /**
         * 流中的每一个元素都会调用这个方法
         */
        // 在 processElement 方法中，
        // 每当收到一条数据（ItemViewCount），
        // 我们就注册一个 windowEnd+1 的定时器
        @Override
        public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
            itemState.add(value);
            // 注册 windowEnd+1 的 EventTime Timer,
            // 当触发时，说明收齐了属于windowEnd窗口的所有商品数据
            ctx.timerService().registerEventTimeTimer(value.windowEnd + 1);
            //registerEventTimeTimer
            //会注册当前key的event time 定时器。
            // 当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
        }

        /**
         * onTimer是一个回调函数，
         * 当注册的定时器触发时调用onTimer，
         * 参数timestamp为定时器所设定的触发的时间戳，
         * out是输出结果集合
         */
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            List<ItemViewCount> allItems = new ArrayList<>();
            for (ItemViewCount item : itemState.get()) {
                allItems.add(item);
            }

            // 提前清除状态中的数据，释放空间
            itemState.clear();

            // 按点击量从大到小排序
            allItems.sort(new Comparator<ItemViewCount>() {
                @Override
                public int compare(ItemViewCount o1, ItemViewCount o2) {
                    return (int) (o2.viewCount - o1.viewCount);
                }
            });

            // 将排名信息格式
            StringBuilder result = new StringBuilder();
            result.append("====================================\n");
            result.append("时间: ").append(new Timestamp(timestamp-1)).append("\n");
            for (int i = 0; i < topSize; i++) {
                ItemViewCount currentItem = allItems.get(i);
                result.append("No").append(i).append(":")
                        .append("  商品ID=").append(currentItem.itemId)
                        .append("  浏览量=").append(currentItem.viewCount)
                        .append("\n");
            }
            result.append("====================================\n\n");

            out.collect(result.toString());
        }

    }

    /** COUNT 统计的聚合函数实现，每出现一条记录加一 */
    public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {
        //AggregateFunction是一个基于中间计算结果状态进行增量计算的函数
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(UserBehavior userBehavior, Long acc) {
            return acc + 1;
        }

        @Override
        public Long getResult(Long acc) {
            return acc;
        }

        @Override
        public Long merge(Long acc1, Long acc2) {
            return acc1 + acc2;
        }
    }

    /** 用于输出窗口的结果 */
    public static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {
        @Override
        public void apply(Tuple key, // 窗口的主键，即 itemId
                          TimeWindow window, // 窗口
                          Iterable<Long> aggregateResult, // 聚合函数的结果，即 count 值
                          Collector<ItemViewCount> out // 输出类型为 ItemViewCount
        ) throws Exception {
            Long itemId = ((Tuple1<Long>)key).f0;
            Long count = aggregateResult.iterator().next();
            out.collect(ItemViewCount.of(itemId, window.getEnd(), count));
        }
    }

    /** 商品点击量(窗口操作的输出类型) */
    public static class ItemViewCount {
        public long itemId;     // 商品ID
        public long windowEnd;  // 窗口结束时间戳
        public long viewCount;  // 商品的点击量

        public static ItemViewCount of(long itemId, long windowEnd, long viewCount) {
            ItemViewCount result = new ItemViewCount();
            result.itemId = itemId;
            result.windowEnd = windowEnd;
            result.viewCount = viewCount;
            return result;
        }
    }
}
